package removeDanglingPage;

import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import mapred.AbstractReducer;
import mapred.util.Bytes;

public class RemoveDP_Reducer extends AbstractReducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>{

	int sum = 0;
	byte[] fx = new byte[16];
	@Override
	public void reduce(BytesWritable key, Iterator<BytesWritable> values,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		
		byte[] p = new byte[16];
		ReadBytes k = new ReadBytes(key);
		p = k.nextUrl();
		k.next();
		int a = k.nextInt();
		if( a < 0) {
			reporter.incrCounter("Remove DP", "lower", 1);
			fx = Arrays.copyOf(p, p.length);
			while(values.hasNext()) {
				sum += -1;
				values.next();
			}
		} 
		
		if( a == 0 ) {
			while(values.hasNext()) {
				reporter.incrCounter("Remove DP", "Removed Edges", 1);
				values.next();
			}
		}
		
		if( a > 0 ) {
			reporter.incrCounter("Remove DP", "higher", 1);
			if(!Arrays.equals(fx, p)) {
				sum = 0;
			}
			sum = sum + a;
			if( sum == 0) {
				reporter.incrCounter("Remove DP", "zero", 1);
				while(values.hasNext()) {
					ReadBytes v = new ReadBytes(values.next());
					byte[] ww = new byte[16];
					ww = v.nextUrl();
					output.collect(new BytesWritable(Bytes.concat(p, Bytes.Integer2Bytes(0))), 
							new BytesWritable(ww));
					byte[] url0 = Bytes.concat(ww, new byte[] {0});
					output.collect(new BytesWritable(Bytes.concat(url0, Bytes.Integer2Bytes(-1))), 
							new BytesWritable(new byte[0]));
				}
			} else {
				while(values.hasNext()) {
					byte[] url1 = Bytes.concat(p, new byte[] {1});
					output.collect(new BytesWritable(Bytes.concat(url1, Bytes.Integer2Bytes(sum))), 
							values.next());
				}
			}
			sum = 0;
		}
	}


//	int sum = 0;
//	
//	@Override
//	public void reduce(Text key, Iterator<Text> values,
//			OutputCollector<Text, Text> output, Reporter reporter)
//			throws IOException {
//		// TODO Auto-generated method stub
//		
//		String keyStr = key.toString();
//		String[] t = keyStr.split(" ");
//		if(Integer.parseInt(t[1]) < 0) {
//			
//			while(values.hasNext()) {
//				sum += Integer.parseInt(t[1]);
//				values.next();
//			
//			}
//			
//		}
//		if(Integer.parseInt(t[1]) == 0) {
//			reporter.incrCounter("Remove DP", "Removed Edges", 1);
//		}
//		if(Integer.parseInt(t[1]) > 0) {
//			reporter.incrCounter("Remove DP", "Sum Edges", 1);
//			if(Integer.parseInt(t[1])+sum == 0) {
//				while(values.hasNext()) {
//					String f = values.next().toString();
//					output.collect(new Text(t[0] + " " +0), new Text(f));
//					output.collect(new Text(f+" "+"-1"), new Text(""));
//				}
//			} else {
//				while(values.hasNext()) {
//					output.collect(new Text(t[0] + " " +(Integer.parseInt(t[1])+sum)), 
//							new Text(values.next().toString()));
//				}
//			}
//			sum = 0;
//			
//		}
//	}

}
